package me.tuyou.wordcount;

import me.tuyou.common.MyOrder;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class TableBatchWordCountTest {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
        DataSource<MyOrder> dataSource = env.fromElements(new MyOrder(1L, "tuyou", 1),
                new MyOrder(2L, "yangrui", 8),
                new MyOrder(3L, "lishunzhen", 3),
                new MyOrder(4L, "tuchuanlin", 5));

        Table allOrder = tableEnv.fromDataSet(dataSource);
        Table filtered = allOrder.where($("amount").isGreaterOrEqual(5));
        DataSet<MyOrder> resultSet = tableEnv.toDataSet(filtered, MyOrder.class);
        resultSet.print();
    }
}
